package com.atguigu.project;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import javax.swing.text.ParagraphView;

/**
 * @author wky
 * @create 2021-07-16-9:25
 */
public class Project_PV {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setParallelism(1);
        //senv.setRuntimeMode(RuntimeExecutionMode.BATCH);
        DataStreamSource<String> streamSource = senv.readTextFile("src/input/UserBehavior.csv");

        SingleOutputStreamOperator<Tuple2<String, Long>> pv = streamSource.process(new ProcessFunction<String, Tuple2<String, Long>>() {
            long count = 0;
            @Override
            public void processElement(String s, Context context, Collector<Tuple2<String, Long>> collector) throws Exception {
                String[] split = s.split(",");
                if ("pv".equals(split[3])) {
                    count++;
                    collector.collect(Tuple2.of("pv", count));
                }
            }
        });
        pv.print();

        senv.execute();

    }
}
